package top.houry.netty.barrage.rocketmq;

import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;
import top.houry.netty.barrage.bo.BarrageMsgBo;
import top.houry.netty.barrage.entity.BarrageMsg;
import top.houry.netty.barrage.service.IBarrageMsgService;
import top.houry.netty.barrage.service.impl.BarrageReceiveMsgServiceImpl;

import java.net.InetAddress;
import java.net.UnknownHostException;

/**
 *
 * RocketMQ
 *  发布订阅 消费者
 *  用于消息转发（集群实现）
 */
@Component
@Slf4j
public class SubConsumerService implements ApplicationContextAware {
    final static String nameServerUrl = "127.0.0.1:9876";
    final static String topic = "distributed";
    final static String ConsumerGROUP = "test-sub-consumer";
    private BarrageReceiveMsgServiceImpl barrageReceiveMsgService;
    @Autowired
    public void setBarrageReceiveMsgService(BarrageReceiveMsgServiceImpl barrageReceiveMsgService) {
        this.barrageReceiveMsgService = barrageReceiveMsgService;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        initConsumer();
    }

    synchronized private void initConsumer() {
        DefaultMQPushConsumer mqPushConsumer = new DefaultMQPushConsumer(ConsumerGROUP);
        mqPushConsumer.setNamesrvAddr(nameServerUrl);
        // 开启广播模式
        // 广播模式下，发送一条消息后，会被当前被广播的所有Consumer消费，但是后面新加入的Consumer不会消费这条消息
        mqPushConsumer.setMessageModel(MessageModel.BROADCASTING);
        // *指定的是tag标签，代表全部消息，不进行任何过滤
        try {
            mqPushConsumer.subscribe(topic,"*");
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        // 注册监听器，进行消息消费
        mqPushConsumer.registerMessageListener((MessageListenerConcurrently) (list, consumeConcurrentlyContext) -> {

            for (MessageExt messageExt : list) {
                String message = new String(messageExt.getBody());
                BarrageMsgBo msg = JSONUtil.toBean(message, BarrageMsgBo.class);
                if (checkNotMySelf(msg)){
                    // 发布的时候，需要把自己的唯一标识给带上
                    // 订阅的时候，如果发现是自己的唯一标识，则不处理，否则进行转发逻辑
                    // 获取自己的ip:port 并从消息里获取唯一标识消息 判断是否一致决定是否转发
                    barrageReceiveMsgService.notifyOtherUser(msg.getVideId(),msg);
                }
                log.info("收到===================== topic:【{}】 的消息【{}】",topic,message);
            }
            log.info("插入数据库成功!");
            // ack确认
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        //启动消费者
        try {
            mqPushConsumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        log.info("test-sub-consumer 消费者已经启动");
    }
    /**
     * 检查当前的消息是不是自己发布的
     * true ： 不是自己发的
     * false : 是自己发的
     */
    private boolean checkNotMySelf(BarrageMsgBo barrageMsgBo) {
        try {
            InetAddress localHost = InetAddress.getLocalHost();
            return !(barrageMsgBo.getIp().equals(localHost.getHostAddress()) && barrageMsgBo.getPort().equals(barrageReceiveMsgService.getNettyConfigProperties().getServerPort()));
        } catch (UnknownHostException e) {
            throw new RuntimeException("获取本地 ip 地址失败");
        }
    }
}
